package org.spbsu.apco.common.serialization;

import org.apache.commons.codec.binary.Base64;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.spbsu.apco.common.Pair;
import org.spbsu.apco.common.data.core.Data;
import org.spbsu.apco.common.monitoring.event.TaskEvent;
import org.spbsu.apco.common.task.core.BaseTask;
import org.spbsu.apco.common.task.core.Ticket;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.*;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.List;

/**
 * User: solozobov
 * Date: 18.11.12
 */

@Component("serializer")
public class Serializer {
    private static final Logger LOG = LoggerFactory.getLogger(Serializer.class);

    {
        Ticket.setSerializer(this);
    }
    @Resource(name = "javaSerializationMechanism")
    private SerializationMechanism serializationMechanism;

    private static class TaskAndInput implements Serializable {
        public final long taskId;
        public final Class taskClass;
        public final byte[] input;

        public TaskAndInput(long taskId, Class taskClass, byte[] input) {
            this.taskId = taskId;
            this.taskClass = taskClass;
            this.input = input;
        }
    }

    public byte[] serializeTaskAndInput(final BaseTask task, final List<?> inputs) {
        final ArrayList<Pair<String,byte[]>> inputsRepresentation = new ArrayList<Pair<String,byte[]>>(inputs.size());
        for (final Object input : inputs) {
            inputsRepresentation.add(serializeObject(input));
        }
        return serializationMechanism.serialize(new TaskAndInput(task.getId(), task.getClass(), serializationMechanism.serialize(inputsRepresentation)));
    }

    public Ticket deserializeTaskAndInput(final byte[] representation) {
        final TaskAndInput taskAndInput = (TaskAndInput) serializationMechanism.deserialize(representation);
        try {
            BaseTask task = null;
            try {
                task = (BaseTask) taskAndInput.taskClass.getConstructor().newInstance();
            } catch (NoSuchMethodException e) {
                LOG.error("Error, you might forgot to implement constructor for " + taskAndInput.taskClass.getSimpleName() + " that has no arguments", e);
            }

            final Field id = BaseTask.class.getDeclaredField("id");
            id.setAccessible(true);
            id.setLong(task, taskAndInput.taskId);

            @SuppressWarnings("unchecked")
            List<Pair<String, byte[]>> inputsRepresentations = (ArrayList<Pair<String, byte[]>>) serializationMechanism.deserialize(taskAndInput.input);

            final List<Object> inputs = new ArrayList<Object>(inputsRepresentations.size());
            for (final Pair<String, byte[]> inputRepresentation : inputsRepresentations) {
                inputs.add(deserializeObject(inputRepresentation));
            }

            return Ticket.of(task, inputs);
        } catch (Exception e) {
            LOG.error("Can't deserialize task and input", e);
            throw new RuntimeException("Can't deserialize task and input", e);
        }
    }

    public byte[] serializeResult(final Object object) {
        return serializationMechanism.serialize(serializeObject(object));
    }

    public Object deserializeResult(final byte[] representation) {
        @SuppressWarnings("unchecked")
        final Pair<String, byte[]> resultRepresentation = (Pair<String, byte[]>) serializationMechanism.deserialize(representation);
        return deserializeObject(resultRepresentation);
    }

    public byte[] encodeTaskEvents(List<TaskEvent> taskEvents) {
        final byte[] serializedEvents = serializationMechanism.serialize(taskEvents);
        return Base64.encodeBase64(serializedEvents);
    }

    public List<TaskEvent> decodeTaskEvents(final byte[] encodedEvent) {
        final byte[] decodedEvents = Base64.decodeBase64(encodedEvent);
        return (List<TaskEvent>) serializationMechanism.deserialize(decodedEvents);
    }


    private Pair<String, byte[]> serializeObject(final Object object) {
        try {
            final byte[] representation;
            if (object instanceof Data) {
                representation = ((Data) object).serialize();
            }
            else if (object instanceof Serializable) {
                representation = serializationMechanism.serialize(object);
            }
            else {
                LOG.error("Object " + object.getClass().getName() + " can't be serialized because it is not instance of Data and not Serializable");
                throw new RuntimeException("Object " + object.getClass().getName() + " can't be serialized because it is not instance of Data and not Serializable");
            }
            return Pair.of(object.getClass().getName(), representation);
        } catch (Exception e) {
            LOG.error("Can't serialize object: " + object.getClass().getSimpleName(), e);
            throw new RuntimeException("Can't serialize object: " + object.getClass().getSimpleName(), e);
        }
    }

    private Object deserializeObject(final Pair<String, byte[]> representation) {
        final String className = representation.first;
        final Class<?> clazz;
        try {
            clazz = Class.forName(className);
        } catch (ClassNotFoundException e) {
            LOG.error("Unknown object class " + className);
            throw new RuntimeException("Unknown object class " + className, e);
        }
        if (Data.class.isAssignableFrom(clazz)) {
            try {
                //noinspection PrimitiveArrayArgumentToVariableArgMethod
                return clazz.getConstructor(byte[].class).newInstance(representation.second);
            } catch (NoSuchMethodException e) {
                LOG.error("Constructor not found for " + className + " that has byte[] argument", e);
                throw new RuntimeException("Constructor not found for " + className + " that has byte[] argument", e);
            } catch (Exception e) {
                LOG.error("Can't deserialize Data object: " + representation.toString(), e);
                throw new RuntimeException("Can't deserialize Data object: " + representation.toString(), e);
            }
        }
        else if (Serializable.class.isAssignableFrom(clazz)) {
            return serializationMechanism.deserialize(representation.second);
        } else {
            LOG.error("Can't deserialize object. It is not instance of Data and not Serializable");
            throw new RuntimeException("Can't deserialize object. It is not instance of Data and not Serializable");
        }
    }

    public void setSerializationMechanism(SerializationMechanism serializationMechanism) {
        this.serializationMechanism = serializationMechanism;
    }
}